package zy.learn.demo.structuredstreaming.join

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/*
输入：
1:
lisi,female,2019-09-16 11:50:00
zs,male,2019-09-16 11:51:00
ww,female,2019-09-16 11:52:00
zhiling,female,2019-09-16 11:53:00
fengjie,female,2019-09-16 11:54:00
yifei,female,2019-09-16 11:55:00
2:
lisi,18,2019-09-16 11:50:00
zs,19,2019-09-16 11:51:00
ww,20,2019-09-16 11:52:00
zhiling,22,2019-09-16 11:53:00
yifei,30,2019-09-16 11:54:00
fengjie,98,2019-09-16 11:55:00

输出
+-------+------+-------------------+---+-------------------+
|   name|   sex|                ts1|age|                ts2|
+-------+------+-------------------+---+-------------------+
|zhiling|female|2019-09-16 11:53:00| 22|2019-09-16 11:53:00|
|     ww|female|2019-09-16 11:52:00| 20|2019-09-16 11:52:00|
|  yifei|female|2019-09-16 11:55:00| 30|2019-09-16 11:54:00|
|     zs|  male|2019-09-16 11:51:00| 19|2019-09-16 11:51:00|
|fengjie|female|2019-09-16 11:54:00| 98|2019-09-16 11:55:00|
|   lisi|female|2019-09-16 11:50:00| 18|2019-09-16 11:50:00|
+-------+------+-------------------+---+-------------------+
 */
object StreamingStaticJoin {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")
    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("StreamingStaticJoin")
      .getOrCreate()

    import spark.implicits._

    // 1. 静态 df
    val arr = Array(("lisi", "male"), ("zhiling", "female"), ("zs", "male"));
    var staticDF = spark.sparkContext.parallelize(arr).toDF("name", "sex")

    // 2. 流式 df
    val lines = spark.readStream
      .format("socket")
      .option("host", "co7-203")
      .option("port", 9999)
      .load()

    val streamDF = lines.as[String].map(line => {
      val arr = line.split(",")
      (arr(0), arr(1).toInt)
    }).toDF("name", "age")

    // 3. join   等值内连接  a.name=b.name
    val resultDF = streamDF.join(staticDF, Seq("name"), "left")

    resultDF.writeStream
      .outputMode("update")
      .format("console")
      .start
      .awaitTermination()
  }
}
